package com.zy.ems.common.mq;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.taobao.metamorphosis.client.extension.spring.MessageBuilder;
import com.taobao.metamorphosis.client.extension.spring.MetaqTemplate;
import com.taobao.metamorphosis.client.producer.SendResult;
import com.zy.ems.common.util.JsonUtils;

/**
 * MQ工具类。
 * 
 * @author Luffy
 * @version
 * @since JDK 1.7
 */
public class MetaQUtils {

    private static final Logger logger = LoggerFactory
            .getLogger(MetaQUtils.class);

    private static MetaQUtils instance = new MetaQUtils();

    private MetaqTemplate metaqTemplate;

    private boolean initialized = false;

    private ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200,
            TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(200));

    private MetaQUtils() {

    }

    protected void init(MetaqUtilInitializer initializer) {
        if (initialized) {
            throw new RuntimeException("缓存环境不能初始化两次!");
        }
        metaqTemplate = initializer.getMetaqTemplate();
        if (metaqTemplate == null) {
            throw new RuntimeException("metaqTemplate不能为空!");
        }

        initialized = true;
    }

    public static MetaQUtils getInstance() {
        return instance;
    }

    public static interface MetaqUtilInitializer {

        /**
         * 初始化过程中，获得默认的缓存集群操作文件
         *
         * @return
         */
        MetaqTemplate getMetaqTemplate();

    }

    /**
     * 启动异步线程发送metaq消息<br/>
     * 
     * @param topic
     * @param msgObj
     * @author sunshine
     * @date 2016年2月29日 下午9:39:55
     */
    public void sendMsg(final String topic, final Object msgObj) {
        // 防止发送空消息
        if (msgObj == null) {
            return;
        }
        final String msg = JsonUtils.writeObject2JSON(msgObj);
        sendMsg(topic, msg);
    }

    /**
     * 启动异步线程发送metaq消息<br/>
     * 
     * @param topic
     * @param msg
     * @author sunshine
     * @date 2016年2月29日 下午9:39:55
     */
    public void sendMsg(final String topic, final String msg) {
        executor.execute(new Runnable() {
            public void run() {
                try {
                    logger.info("topic start to send ：" + topic + " message : "
                            + msg);
                    SendResult result = metaqTemplate.send(
                            MessageBuilder.withTopic(topic).withBody(msg));
                    logger.info("topic info ：" + topic + " has send " + msg
                            + " to metaq " + result.getPartition());
                }
                catch (InterruptedException e) {
                    logger.error(e.getMessage(), e);
                }
            }
        });
    }
}
